package day05;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Description:  迟到数据发送到输出流中
 * @Author: ZYX
 * @Date: 2022/2/14 10:08
 * @Version: 1.0
 */
public class Demo02 {

    // 定义侧输出流的名字：侧输出标签
    private static OutputTag<String> lateElement = new OutputTag<String>("late-element"){};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<String> result = env.addSource(new SourceFunction<Tuple2<String, Long>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Long>> sourceContext) throws Exception {
                // 指定时间戳发送数据
                sourceContext.collectWithTimestamp(Tuple2.of("hello world", 1000L), 1000L);
                // 发送水位线
                sourceContext.emitWatermark(new Watermark(999L));

                sourceContext.collectWithTimestamp(Tuple2.of("hell flink", 1000L), 1000L);
                sourceContext.emitWatermark(new Watermark(1999L));
                sourceContext.collectWithTimestamp(Tuple2.of("late data", 1000L), 1000L);
            }

            @Override
            public void cancel() {

            }
        })
                .process(new ProcessFunction<Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> value, Context context, Collector<String> out) throws Exception {
                        if (value.f1 < context.timerService().currentWatermark()) {
                            context.output(lateElement, "迟到元素发送到侧输出流" + value);
                        } else {
                            out.collect("正常元素到达：" + value);
                        }
                    }
                });
        result.print();

        result.getSideOutput(lateElement).print("侧输出");

        env.execute();
    }

}
